死信|都会_rabbitmq详解

篇首语：本文由编程笔记#小编为大家整理，主要介绍了rabbitmq详解相关的知识，希望对你有一定的参考价值。

rabbitmq

• 一、简介
• 二、业务场景
• • 1、异步
  • 2、应用解耦
  • 3、流量削峰
  
  
• 三、下载
• 四、界面认识
• 五、五种模型示例
• • 0、springboot依赖配置
  • 1、Hello World简单模型
  • 2、Work queues工作队列
  • 3、Publish/Subscribe发布订阅模型
  • 4、Routing路由模型
  • 5、Topics主题模型
  • 6、消息转换器
  
  
• 六、进阶
• • 1、基于插件延迟队列
  • 2、TTL队列
  • 3、死信队列
  • 4、消息确认
  • • 1、发送消息确认机制
    • 2、消费者消息确认机制
    
    
  
  
• 七、rabbitmq集群搭建
• • 1、普通集群
  • 2、镜像集群&＃xff08;高可用&＃xff09;&＃xff08;推荐&＃xff09;
  
  
• 八、与其他mq的区别
• 九、rabbitmq常见面试题

RabbitMQ是一个实现了AMQP&＃xff08;Advanced Message Queuing Protocol&＃xff09;高级消息队列协议的消息队列服务&＃xff0c;用Erlang语言。是面向消息的中间件。

你可以把它想像成一个邮局&＃xff1a;你把信件放入邮箱&＃xff0c;邮递员就会把信件投递到你的收件人处。在这个比喻中&＃xff0c;RabbitMQ是一个邮箱、邮局、邮递员。RabbitMQ和邮局的主要区别是&＃xff0c;它处理的不是纸&＃xff0c;而是接收、存储和发送二进制的数据——消息。

主要流程&＃xff1a;生产者&＃xff08;Producer&＃xff09;与消费者&＃xff08;Consumer&＃xff09;和 RabbitMQ 服务&＃xff08;Broker&＃xff09;建立连接&＃xff0c; 然后生产者发布消息&＃xff08;Message&＃xff09;同时需要携带交换机&＃xff08;Exchange&＃xff09; 名称以及路由规则&＃xff08;Routing Key&＃xff09;&＃xff0c;这样消息会到达指定的交换机&＃xff0c;然后交换机根据路由规则匹配对应的 Binding&＃xff0c;最终将消息发送到匹配的消息队列&＃xff08;Quene&＃xff09;&＃xff0c;最后 RabbitMQ 服务将队列中的消息投递给订阅了该队列的消费者&＃xff08;消费者也可以主动拉取消息&＃xff09;。

1、异步

如&＃xff1a; 用户注册发送&＃xff0c;注册邮件、注册短信&＃xff0c;
传统做法&＃xff1a;

1、串行 &＃xff08;先发送邮件、再发短信&＃xff09;。问题&＃xff1a;持续时间长

2、并行&＃xff08;将注册信息写入数据库后&＃xff0c;同时发送邮件、短信&＃xff09;&＃xff0c;速度快、但不能满足高吞吐需求。

消息队列做法&＃xff1a;
将数据写入数据库、同时发送消息给发送邮件和注册&＃xff0c;异步处理

2、应用解耦

如&＃xff1a;双十一购物节&＃xff0c;用户下单后、订单系统通知库存系统。

传统做法:
订单系统调用库存系统接口。问题&＃xff1a;库存接口故障&＃xff0c;订单就会失败&＃xff0c;而损失大量订单

消息队列做法

订单系统&＃xff1a;下单&＃xff0c;订单系统完成持久化&＃xff0c;将消息写入队列&＃xff0c;返回下单成功给用户
库存系统&＃xff1a;订阅下单的消息&＃xff0c;获取下单消息&＃xff0c;进行库操作&＃xff0c;就算库存系统故障&＃xff0c;消息队列也能保证消息可靠投递&＃xff0c;不会导致消息丢失。

3、流量削峰

如&＃xff1a;秒杀活动、一般会因为流量过大&＃xff0c;导致应用挂掉&＃xff0c;一般在应用前端加入消息队列。

作用&＃xff1a;1、可以控制活动人数&＃xff0c;超过一定阈值&＃xff0c;订单直接丢弃
2、可以缓解短时间的高流量压垮应用&＃xff08;应用程序按自己的最大处理能力获取订单&＃xff09;

消息队列做法
1、用户的请求&＃xff0c;服务器收到后&＃xff0c;首先写入消息队列&＃xff0c;加入消息队列长度最大值&＃xff0c;则直接抛弃用户请求或跳转到错误页面
2、秒杀业务根据消息队列中的请求信息&＃xff0c;再做后续处理

1、docker 安装 rabbitmq

docker pull rabbitmq:3.7.7-management


2、​​启动镜像&＃xff08;用户名和密码设置为 guest guest&＃xff09;

docker run -dit --name rabbitmq3.7.7 -e RABBITMQ_DEFAULT_USER&＃61;guest -e RABBITMQ_DEFAULT_PASS&＃61;guest -v /home/rabbitmq/data:/var/lib/rabbitmq -p 15672:15672 -p 5672:5672 rabbitmq:3.7.7-management


3、访问 rabbitmq 管理界面

http://127.0.0.1:15672 账号密码都是 guest

4、docker 安装 rabbitMQ 延时队列插件&＃xff08;delayed_message_exchange&＃xff09;

下载解压文件 链接&＃xff1a;https://pan.baidu.com/s/1PpeOn8NJT4hgh7ZBP0J0OA?pwd&＃61;u2gu
提取码&＃xff1a;u2gu

拷贝插件文件到 rabbitMQ 的 Docker 容器中
先解压

unzip unzip rabbitmq_delayed_message_exchange-20171201-3.7.x.zip


拷贝插件

docker cp rabbitmq_delayed_message_exchange-20171201-3.7.x.ez rabbitmq3.7.7:/plugins


进入容器&＃xff1a;

docker ps // 查看启动容器信息
docker exec -it 镜像ID /bin/bash //开启进入终端


查看插件列表

rabbitmq-plugins list


启用插件

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

四、界面认识

1、概要

2、连接

3、通道

4、交换机

交换器&＃xff0c;用来接收生产者发送的消息并将这些消息路由给服务器中的队列

5、队列

点击名称进去&＃xff0c;可以看到队列的详细信息

get Message可以看到消息的内容

arguments具体参数如下:

6、用户

就是添加用户和设置用户权限

0、springboot依赖配置

依赖


<dependency>
<groupId>org.springframework.bootgroupId>
<artifactId>spring-boot-starter-amqpartifactId>
dependency>


yml配置

spring:
application:
name: rabbitmq
rabbitmq:
host: 127.0.0.1
port: 5672
username: guest
password: guest
virtual-host: /


1、Hello World简单模型

一对一消费&＃xff0c;只有一个消费者能接收到

消费者

&＃64;Component
public class HolloWordListener
// &＃64;RabbitListener(queues &＃61; ("simple.queue")) // queues需手动先创建队列
&＃64;RabbitListener(queuesToDeclare &＃61; &＃64;Queue("simple.queue")) // queuesToDeclare 自动声明队列
public void holloWordListener(String message)
System.out.println("message &＃61; " &＃43; message);




生产者

&＃64;Autowired
private RabbitTemplate rabbitTemplate;
&＃64;Test
public void testSimpleQueue()
String queueName &＃61; "simple.queue"; // 队列名称
String message &＃61; "heel&＃xff0c;simple.queue"; // 要发送的消息
rabbitTemplate.convertAndSend(queueName,message);



2、Work queues工作队列

多个消费者&＃xff0c;你一个我一个分配消费消息&＃xff0c;有预取机制&＃xff0c;默认公平消费&＃xff0c;可配置能者多劳模式&＃xff0c;谁完成的快&＃xff0c;谁多做一点

消费者

&＃64;Component
public class WoekWordListener
&＃64;RabbitListener(queuesToDeclare &＃61; &＃64;Queue("workQueue")) // queuesToDeclare 自动声明队列
public void holloWordListener(String message) throws InterruptedException
Thread.sleep(200);
System.out.println("message1 &＃61; " &＃43; message);


&＃64;RabbitListener(queuesToDeclare &＃61; &＃64;Queue("workQueue")) // queuesToDeclare 自动声明队列
public void holloWordListener1(String message) throws InterruptedException
Thread.sleep(400);
System.out.println("message2 &＃61; " &＃43; message);




生产者

&＃64;Autowired
private RabbitTemplate rabbitTemplate;

&＃64;Test
public void testWorkQueue()
String queueName &＃61; "workQueue";
String message &＃61; "hello&＃xff0c;work.queue__";
for (int i &＃61; 0; i < 10; i&＃43;&＃43;)
rabbitTemplate.convertAndSend(queueName,message&＃43;i);
System.out.println("i &＃61; " &＃43; i);




取消预取机制&＃xff0c;能者多劳配置

spring:
rabbitmq:
host: 127.0.0.1
port: 5672
username: guest
password: guest
virtual-host: /
listener:
simple:
prefetch: 1 # 每次只能获取一条&＃xff0c;处理完成才能获取下一条


3、Publish/Subscribe发布订阅模型

发布订阅模式与之前案例的区别就是允许将同一消息发送给多个消费者。
实现方式是加入了exchange(交换机)&＃xff0c;注意&＃xff1a;交换机是不缓存消息的

使用fanout交换机&＃xff0c;会将接收到的消息路由到每一个跟其绑定的queue(队列)

消费者

// 消费者直接绑定交换机&＃xff0c;指定类型为fanout
&＃64;Component
public class FanoutExchangeListener
// 不指定队列&＃xff0c;消息过了就没了
// &＃64;RabbitListener(bindings &＃61; &＃64;QueueBinding(value &＃61; &＃64;Queue,exchange &＃61; &＃64;Exchange(value &＃61; "fanoutTest",type &＃61; ExchangeTypes.FANOUT)))
// 指定队列&＃xff0c;可以接收缓存到队列里的消息
&＃64;RabbitListener(bindings &＃61; &＃64;QueueBinding(value &＃61; &＃64;Queue(value &＃61;"test",durable &＃61; "true" ),exchange &＃61; &＃64;Exchange(value &＃61; "fanoutTest",type &＃61; ExchangeTypes.FANOUT)))
public void reveivel(String message)
System.out.println("message &＃61; " &＃43; message);

&＃64;RabbitListener(bindings &＃61; &＃64;QueueBinding(value &＃61; &＃64;Queue,exchange &＃61; &＃64;Exchange(value &＃61; "fanoutTest",type &＃61; ExchangeTypes.FANOUT)))
public void reveivel2(String message)
System.out.println("message1 &＃61; " &＃43; message);




生产者

&＃64;Autowired
private RabbitTemplate rabbitTemplate;

&＃64;Test
public void tesyPubSubQueue()
// 参数1&＃xff1a;交换机名称 &＃xff0c; 参数2routingKey,&＃xff08;fanout类型可不写&＃xff09; , 参数3&＃xff0c;消息内容
rabbitTemplate.convertAndSend("fanoutTest","","消息内容");



4、Routing路由模型

routing模型也是将消息发送到交换机

使用的是Direct类型的交换机&＃xff0c;会将接收到的消息根据规则路由到指定的Queue(队列)&＃xff0c;因此称为路由模式

消费者

// 消费者直接绑定交换机&＃xff0c;指定类型为direct&＃xff0c;并指定key表示能消费的key
&＃64;Component
public class RoutingExchangeListener
// 不指定队列&＃xff0c;消息过了就没了
// &＃64;RabbitListener(bindings &＃61; &＃64;QueueBinding(value &＃61; &＃64;Queue,exchange &＃61; &＃64;Exchange(value &＃61; "direstTest",type &＃61; ExchangeTypes.DIRECT),key &＃61; "info","error"))

// 指定队列&＃xff0c;可以接收缓存到队列里的消息
// key &＃61; "info","error" 表示我能接收到routingKey为 info和error的消息
&＃64;RabbitListener(bindings &＃61; &＃64;QueueBinding(value &＃61; &＃64;Queue(value &＃61;"test1",durable &＃61; "true" ),exchange &＃61; &＃64;Exchange(value &＃61; "direstTest",type &＃61; ExchangeTypes.DIRECT),key &＃61; "info","error"))
public void receivel(String message)
System.out.println("message &＃61; " &＃43; message);

// key &＃61; "error" 表示我只能接收到routingKey为 error的消息
&＃64;RabbitListener(bindings &＃61; &＃64;QueueBinding(value &＃61; &＃64;Queue,exchange &＃61; &＃64;Exchange(value &＃61; "direstTest",type &＃61; ExchangeTypes.DIRECT),key &＃61; "error"))
public void receivel1(String message)
System.out.println("message1 &＃61; " &＃43; message);




生产者

&＃64;Autowired
private RabbitTemplate rabbitTemplate;

// 路由模型
&＃64;Test
public void direstExchangeTest()
rabbitTemplate.convertAndSend("direstTest","info","发送info的key的路由消息");

// 路由模型
&＃64;Test
public void direstExchangeTest1()
rabbitTemplate.convertAndSend("direstTest","error","发送error的key的路由消息");



5、Topics主题模型

topicExchange与directExchange类型&＃xff0c;区别在于routingKey必须是多个单词的列表&＃xff0c;并且以 . 分隔

*&＃xff08;代表通配符&＃xff0c;任意一个字段&＃xff09;
#&＃xff08;号代表一个或多个字段)

消费者

&＃64;Component
public class TopicsExchangeListener
// 不指定队列&＃xff0c;消息过了就没了
// &＃64;RabbitListener(bindings &＃61; &＃64;QueueBinding(value &＃61; &＃64;Queue,exchange &＃61; &＃64;Exchange(name &＃61; "topicList",type &＃61; ExchangeTypes.TOPIC),key &＃61; "user.save","user.*"))

// 指定队列&＃xff0c;可以接收缓存到队列里的消息
// key &＃61; "user.save","user.*" 表示能消费 routingkey为 user.save 和 user.任意一个字符 的消息
&＃64;RabbitListener(bindings &＃61; &＃64;QueueBinding(value &＃61; &＃64;Queue(value &＃61;"test2",durable &＃61; "true" ),exchange &＃61; &＃64;Exchange(name &＃61; "topicList",type &＃61; ExchangeTypes.TOPIC),key &＃61; "user.save","user.*"))
public void recevicel(String message)
System.out.println("message &＃61; " &＃43; message);

// key &＃61; "order.#","user.*" 表示能消费 routingkey为 order.一个或多个字符 和 user.任意一个字符 的消息
&＃64;RabbitListener(bindings &＃61; &＃64;QueueBinding(value &＃61; &＃64;Queue,exchange &＃61; &＃64;Exchange(name &＃61; "topicList",type &＃61; ExchangeTypes.TOPIC),key &＃61; "order.#","user.*"))
public void recevicel1(String message)
System.out.println("message1 &＃61; " &＃43; message);




生产者

&＃64;Autowired
private RabbitTemplate rabbitTemplate;

&＃64;Test
public void topicTest()
rabbitTemplate.convertAndSend("topicTest","user.save","topic路由消息,use.save");


&＃64;Test
public void topicTest1()
rabbitTemplate.convertAndSend("topicTest","order.select.getone","topic路由消息,order.select.getone");



6、消息转换器

代码里直接发送对象&＃xff0c;虽然接收的到消息&＃xff0c;但是rabbitmq的界面上看到的消息会是乱码

依赖

<dependency>
<groupId>com.fasterxml.jackson.dataformatgroupId>
<artifactId>jackson-dataformat-xmlartifactId>
<version>2.9.10version>
dependency>


配置

&＃64;Configuration
public class rabbitmqConfig
// 消息转换配置
&＃64;Bean
public MessageConverter jsonMessageConverter()
return new Jackson2JsonMessageConverter();




再次发送就会是转换好的消息

1、基于插件延迟队列

延迟队列非常常用且好用&＃xff0c;可以将消息发送后使消费者延迟接收。

RabbitAdmin配置

RabbitAdmin是用于对交换机和队列进行管理&＃xff0c;用于创建、绑定、删除队列与交换机&＃xff0c;发送消息的组件。

import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
&＃64;Configuration
public class RabbitAdminConfig
&＃64;Value("$spring.rabbitmq.host")
private String host;
&＃64;Value("$spring.rabbitmq.username")
private String username;
&＃64;Value("$spring.rabbitmq.password")
private String password;
&＃64;Value("$spring.rabbitmq.virtualhost")
private String virtualhost;
&＃64;Bean
public ConnectionFactory connectionFactory()
CachingConnectionFactory connectionFactory &＃61; new CachingConnectionFactory();
connectionFactory.setAddresses(host);
connectionFactory.setUsername(username);
connectionFactory.setPassword(password);
connectionFactory.setVirtualHost(virtualhost);
return connectionFactory;

&＃64;Bean
public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory)
RabbitAdmin rabbitAdmin &＃61; new RabbitAdmin(connectionFactory);
rabbitAdmin.setAutoStartup(true);
return rabbitAdmin;




封装发送延迟队列工具类

import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.CustomExchange;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;
&＃64;Component
public class DelayedQueue
// routingKey
private static final String DELAYED_ROUTING_KEY &＃61; "delayed.routingkey";
// 延迟队列交换机
private static final String DELAYED_EXCHANGE &＃61; "delayed.exchange";
&＃64;Autowired
RabbitTemplate rabbitTemplate;
&＃64;Resource
RabbitAdmin rabbitAdmin;
/**
* 发送延迟队列
* &＃64;param queueName 队列名称
* &＃64;param params 消息内容
* &＃64;param expiration 延迟时间 毫秒
*/
public void sendDelayedQueue(String queueName, Object params, Integer expiration)
// 先创建一个队列
Queue queue &＃61; new Queue(queueName);
rabbitAdmin.declareQueue(queue);
// 创建延迟队列交换机
CustomExchange customExchange &＃61; createCustomExchange();
rabbitAdmin.declareExchange(customExchange);
// 将队列和交换机绑定
Binding binding &